package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object Demo4Tran {
  def main(args: Array[String]): Unit = {
    /**
      * 1、创建生产者
      *
      */

    val properties = new Properties
    //指定kafka链接参数

    //指定kafka briker列表
    properties.setProperty("bootstrap.servers", "master:9092")

    //指定key value序列化类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("transactional.id", "123")

    val kafkaProducer = new KafkaProducer[String, String](properties)

    /**
      * kafka事务
      *
      */

    //开始事务
    kafkaProducer.initTransactions()
    kafkaProducer.beginTransaction()


    //发送数据到kafka
    println("发送第一条数据")
    kafkaProducer.send(new ProducerRecord[String, String]("tran", "java"))

    Thread.sleep(5000)

    println("发送第二条数据")
    kafkaProducer.send(new ProducerRecord[String, String]("tran", "scala"))

    //提交事务
    kafkaProducer.commitTransaction()


    kafkaProducer.close()

    /**
      * --isolation-level read_committed : 读已提交的数据
      * kafka-console-consumer.sh --bootstrap-server  master:9092 --from-beginning --isolation-level read_committed --topic tran
      *
      */
  }

}
